package com.message.queue.rocket.component;

import com.message.queue.constant.RocketConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.*;

/**
 * 封装rocket生产者对象
 * Created by 张波 on 2019/10/19.
 */
@Slf4j
@Component
public class RocketProduct implements InitializingBean {

    @Autowired
    private TransactionListenerImpl transactionListener;

    private TransactionMQProducer product;

    private ExecutorService executorService;

    public static final String PRODUCER_GROUP = "trans_product_group";

    /**
     * 对象构建
     */
    private RocketProduct() {
        this.product = new TransactionMQProducer(PRODUCER_GROUP);
        this.executorService = new ThreadPoolExecutor(5, 100, 30, TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                /*创建回调方法线程*/
                Thread thread = new Thread(r);
                thread.setName(PRODUCER_GROUP + "- thread_check");
                return thread;
            }
        });
        product.setNamesrvAddr(RocketConstant.NAME_SERVER);
        product.setExecutorService(executorService);
    }

    /**
     * 类加载完成后，初始化监听器，同时启动生产者对象
     * @throws Exception
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        this.product.setTransactionListener(transactionListener);
        this.start();
    }

    private void start() {
        try {
            this.product.start();
        } catch (MQClientException e) {
            log.error("MQ启动失败：{}", e);
        }
    }

    /**
     * 关闭消息队列生产者
     */
    public void shutdown() {
        this.product.shutdown();
    }

    /**
     * 发送事物消息
     * @param message
     * @param args
     * @return
     */
    public TransactionSendResult sendMessage(Message message, Object args) {
        TransactionSendResult transactionSendResult = null;
        try {
            transactionSendResult = product.sendMessageInTransaction(message, args);
        } catch (MQClientException e) {
            log.error("发送消息异常：{}", e);
        }
        return transactionSendResult;
    }
}
